from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import KafkaError
import json

# Kafka集群地址
BOOTSTRAP_SERVERS = [
    '10.76.123.10:6667',
    '10.76.123.11:6667',
    '10.76.123.12:6667',
    '10.76.123.13:6667',
    '10.76.123.14:6667',
    '10.76.123.15:6667'
]

# 示例主题
TEST_TOPIC = 'test_topic'


def produce_messages():
    """创建Kafka生产者并发送消息"""
    # 配置生产者
    producer = KafkaProducer(
        bootstrap_servers=BOOTSTRAP_SERVERS,
        value_serializer=lambda v: json.dumps(v).encode('utf-8'),  # 序列化消息为JSON
        retries=5  # 重试次数
    )

    try:
        # 发送消息
        for i in range(10):
            message = {'key': i, 'value': f'message_{i}'}
            future = producer.send(TEST_TOPIC, message)
            # 同步等待确认（可选）
            record_metadata = future.get(timeout=10)
            print(f"消息发送成功: 主题={record_metadata.topic}, 分区={record_metadata.partition}, 偏移量={record_metadata.offset}")
    except KafkaError as e:
        print(f"发送消息失败: {e}")
    finally:
        # 确保所有消息都被发送
        producer.flush()
        # 关闭连接
        producer.close()


def consume_messages():
    """创建Kafka消费者并接收消息"""
    # 配置消费者
    consumer = KafkaConsumer(
        TEST_TOPIC,
        bootstrap_servers=BOOTSTRAP_SERVERS,
        auto_offset_reset='earliest',  # 从最早的消息开始消费
        value_deserializer=lambda m: json.loads(m.decode('utf-8'))  # 反序列化JSON消息
    )

    try:
        # 消费消息
        for message in consumer:
            print(f"接收到消息: 主题={message.topic}, 分区={message.partition}, 偏移量={message.offset}, 键={message.key}, 值={message.value}")
    except KeyboardInterrupt:
        print("手动停止消费者")
    finally:
        # 关闭连接
        consumer.close()


if __name__ == "__main__":
    print("===== 开始发送消息 =====")
    # produce_messages()

    print("\n===== 开始接收消息 =====")
    consume_messages()
